package demo.mqserver.datacenter;

import demo.common.BinaryTool;
import demo.common.MqException;
import demo.mqserver.core.MSGQueue;
import demo.mqserver.core.Message;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

//消息文件所在的目录 和 文件名
@Slf4j
@Data
public class MessageFileManager {

    //定义一个内部类，表示队列的统计信息
    //优先考虑使用 static
    static public class Stat{
        public int totalCount;  //总消息量
        public int validCount;  //有效消息量
    }

    public void init(){
        //暂时不做额外的初始化工作，后续扩展
    }


    //获取指定队列的消息所在的路径
    private String getQueueDir(String queueName){
        return "./data/" + queueName;
    }

    //获取队列的消息数据文件路径
    //二进制文件一般使用 .bin  .dat 后缀
    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //获取队列的消息统计文件路径
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/queue_stat.txt";
    }



    private Stat readStat(String queueName){
        //消息统计文件是文本文件，可以直接使用 Scanner 来读取文件内容
        Stat stat = new Stat();
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName, Stat stat){
        //使用 printWriter 写文件
        //OutputStream 打开文件，默认情况下会直接把原文件清空（相当于是覆盖），可以添加 true（追加）
        try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        //1.创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //不存在，创建
            boolean ok = baseDir.mkdirs();
            if(!ok){
                throw new IOException("创建目录失败！baseDir = " + baseDir.getAbsolutePath());
            }
        }
        //2.创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            boolean ok = queueDataFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！queueDataFile = " + queueDataFile.getAbsolutePath());
            }
        }
        //3.创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean ok = queueStatFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！queueStatFile = " + queueStatFile.getAbsolutePath());
            }
        }
        //4.初始消息统计文件  0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    //删除队列的目录和文件
    //队列删除后，对应的消息文件等也要删除
    public void destroyQueueFiles(String queueName) throws IOException {
        //删除文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        //删除目录
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 || !ok2 || !ok3){
            //任意一个删除失败，就算整体删除失败
            throw new IOException("删除队列目录和文件失败! baseDir =" + baseDir.getAbsolutePath());
        }
    }

    //检查队列的目录和文件是否存在
    //后续生产者给 broker server 生产消息，这个消息可能需要记录到文件上（取决于消息是否需要持久化 isValid）
    public boolean checkFilesExits(String queueName){
        //判断队列的数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }


    //把一个新的消息，放到对应的文件中
    //queue 表示要把消息写入的队列     message 表示要写的消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.检查当前要写入的队列对应的文件是否存在
        if(!checkFilesExits(queue.getName())){
            throw new MqException("[MessageFileManager] 队列对应的文件不存在！queueName = " + queue.getName());
        }
        //2.把 message 对象序列化，转成二进制的字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue){
            //3.获取当前队列的长度，计算 Message 对象的 offsetBeg 和 offsetEnd
            //把新的 message 数据写入到队列数据文件的末尾，Message 对象的 offsetBen 就是文件长度 + 4
            //offsetEnd 就是文件长度 + ￥ + Message 自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            //queueDataFile.length() 获取文件长度（单位字节）
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            //4.写入消息到数据文件（追加到文件末尾）
            try(OutputStream outputStream = new FileOutputStream(queueDataFile, true)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    //写当前信息的长度，占据 4 个 字节
                    dataOutputStream.writeInt((messageBinary.length));
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            //5，更新信息统计文件（isValid）
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }


    //删除消息
    //逻辑删除，把硬盘长存出的数据中的 isValid 属性设置为 0
    //1.把文件中的这一段数据读出来，还原成 Message 对象
    //2.把 isValid 改成 0
    //3.把上述数据重新写回文件
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        //Message 对象要包含有效的 offsetBeg 和 offsetEnd
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                //1.从文件中读取对应的 Message 数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                //2.把二进制数据转换成 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //3.把 isValid 设置成无效
                diskMessage.setIsValid((byte) 0x0);
                //此处非必须修改这个参数，isValid 代表的是内存中管理的 Message 对象
                //这个对象也即将要销毁了
                //4.写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                //上面seek过了，但是读取文件会使光标移动（覆写）
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
                //只有一个字节发生改变
                //5.更新统计文件
                Stat stat = readStat(queue.getName());
                if (stat.validCount > 0) {
                    stat.validCount -= 1;
                }
                writeStat(queue.getName(), stat);
            }
        }
    }

    //从文件中读取所有的消息内容，加载到内存中（具体是放到一个链表里）
    //在程序启动时，进行调用
    //LinkedList 后续进行头删操作
    //参数只是一个 queueName，该方法不需要加锁
    //程序启动时调用，服务器此时还不能处理请求
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
                //记录当前文件的光标
                long currentOffset = 0;
                //一个文件中包含了很多消息，循环读取
                while(true){
                     //1.读取当前消息长度
                    //readInt 读到文件末尾会抛出 EOFException 异常
                    int messageSize = dataInputStream.readInt();
                    //2.按照上述长度，读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if(messageSize != actualSize){
                        //如果不匹配，说明文件有问题，抛异常
                        throw new MqException("[MessageFileManager]！文件格式错误！queueName = " + queueName);
                    }
                    //3.把读取到的二进制数据，反序列化返回 Message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //4.判断是否是无效对象
                    if(message.getIsValid() != 0x1){
                        //更新 offset
                        currentOffset += (4 + messageSize);
                        //无效数据直接跳过
                        continue;
                    }
                    //5.有效数据，把 Message 对象加入到链表中，加入前填写 offsetBeg 和 offsetEnd
                    //计算 offset 时，需要知道当前下标的位置，DataInputStream 不方便获取下标
                    //因此手动计算文件下标
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            }catch (EOFException e){
                //catch 不是处理“异常”，而是在"正常"处理业务逻辑，文件读到末尾会被 readInt 抛出异常
                log.info("[MessageFileManager] 恢复 Message 数据完成！");
            }
        }
        return messages;
    }


    //判断是否针对当前队列的消息数据文件进行GC
    public boolean checkGC(String queueName){
        //判定是否要 GC，根据总消息数和有效消息数，两个值都在 消息统计文件 中
        Stat stat = readStat(queueName);
        if(stat.totalCount > 2000 && (double) stat.validCount / (double) stat.validCount < 0.5){
            return true;
        }
        return false;
    }

    //GC 后文件暂时存储的位置，再删除原 queue_data.txt 文件后，成为新的 queue_data.txt
    private String getQueueDataNewFile(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    //真正执行消息数据文件的垃圾回收操作，使用复制算法
    //创建一个新的文件 queue_data_new.txt，把之前的有效消息都读取出来，存到新文件里
    //删除旧的文件，把新文件改名回 queue_data.txt
    //更新消息统计文件
    public void gc(MSGQueue queue) throws IOException, MqException, ClassNotFoundException {
        //进行 GC ，其他线程不能针对队列的消息文件做任何修改
        synchronized (queue){
            //GC 比较耗时，统计消耗时间
            long gcBeg = System.currentTimeMillis();

            //1.创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewFile(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况，这个文件不应该存在。如果存在，说明上次 gc 时，程序崩溃了
                throw new MqException("[MessageDataManager] gc 时发现 queue_data_new.txt 文件已经存在！queueName = "+ queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageDataManager] 创建文件失败！queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }
            //2.从旧的文件中，读取所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            //3.把有效消息写入到新的文件中
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        //四字节的消息长度
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //4.删除旧文件，命名新文件
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageDataManager] 删除旧的数据文件失败！queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            //重命名
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageDataManager] 文件重命名失败！queueDatNewFile = " + queueDataNewFile.getAbsolutePath() + "，queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            //5.更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            long gcEnd = System.currentTimeMillis();
            log.info("[MessageManager] gc 执行完毕！queue.getName = " + queue.getName()+ "，time = " + (gcEnd - gcBeg) + "ms");
        }
    }
}
